package org.ala.tcp_feign_common.serialize.kryo;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.NoSuchElementException;

import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

/**
 * kryo池子
 *
 * @author 骆毅(ala)
 * @date 2024年4月19日
 */
public class KryoPools {

	/**
	 * idle
	 */
	public static int IDLE = 4;
	/**
	 * max
	 */
	public static int MAX = 10;
	
	protected ObjectPool<Kryo> kryoPool;

	public void init(int idle, int max) {
		//	初始化对象池
		GenericObjectPoolConfig<Kryo> config = new GenericObjectPoolConfig<Kryo>();  
        config.setMaxTotal(max);  
        config.setMaxIdle(idle);  
        config.setMinIdle(0);  
        config.setTestOnBorrow(true);  
        config.setTestOnReturn(true);  
        config.setTestWhileIdle(true);  
        config.setNumTestsPerEvictionRun(3);  
        config.setBlockWhenExhausted(false);  
        config.setJmxEnabled(true);  
        //	对象创建工厂
        PooledObjectFactory<Kryo> factory = new BasePooledObjectFactory<Kryo>() {
			@Override
			public Kryo create() throws Exception {
				Kryo kryo = new Kryo();
				kryo.setRegistrationRequired(false);
				kryo.setReferences(true);
				return kryo;
			}
			@Override
			public PooledObject<Kryo> wrap(Kryo obj) {
				return new DefaultPooledObject<Kryo>(obj);
			}
        };
        kryoPool = new GenericObjectPool<Kryo>(factory, config);  		
	}
	
	
	
	/**
	 * 在kryo作用域内工作
	 * @throws Exception 
	 * @throws IllegalStateException 
	 * @throws NoSuchElementException 
	 */
	public <T> T withKryo(ICallable<T> callable) throws NoSuchElementException {
		Kryo kryo = null;
		int times = 0;
		//	重试3次，每次休眠1s
		while (kryo == null && times++ < 3) {
			try {
				kryo = kryoPool.borrowObject();
			} catch (Exception e) {
				//	休眠2s重试
				synchronized (this) {
					try {
						this.wait(1000l);
					} catch (InterruptedException e1) {
					}
				}
			}
		}
		if (kryo == null) {
			throw new NoSuchElementException();
		}
		
		try {
			return callable.call(kryo);
		} finally {
			try {
				kryoPool.returnObject(kryo);
			} catch (Exception e) {
			}
		}
	}
	public static interface ICallable<T> {
		T call(Kryo kryo);
	}
	
	
	/**
	 * 序列化
	 */
	public byte[] serialize(Object obj) {
		return withKryo(kryo -> {
			ByteArrayOutputStream outStream =  new  ByteArrayOutputStream();
			Output out = new Output(outStream, 1024);
			try {
				kryo.writeObject(out, obj);
				out.flush();
				return outStream.toByteArray();
			} finally {
				out.close();
			}
		});
	}
	/**
	 * 反序列化
	 */
	public <T> T deserialize(byte[] bytes, Class<T> clazz) {
		return withKryo(kryo -> {
			ByteArrayInputStream inStream =  new  ByteArrayInputStream(bytes);
			Input in = new Input(inStream, 1024);
			try {
				return (T)kryo.readObject(in, clazz);
			} finally {
				in.close();
			}
		});
	}
	
	
}
